package com.wxzhou.wangzhuan.service;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.wxzhou.wangzhuan.constants.CacheNames;
import com.wxzhou.wangzhuan.entity.UserWithdrawal;
import io.quarkus.redis.datasource.ReactiveRedisDataSource;
import io.quarkus.redis.datasource.pubsub.ReactivePubSubCommands;
import io.smallrye.mutiny.Uni;
import jakarta.enterprise.context.ApplicationScoped;
import jakarta.inject.Inject;
import org.jboss.logging.Logger;

import java.util.List;

/**
 * redis订阅发布服务
 * 用于发布同步任务是否完成的状态
 */
@ApplicationScoped
public class WithdrawRedisPubService {

    private static final Logger LOG = Logger.getLogger(WithdrawRedisPubService.class);
    private final ReactivePubSubCommands<String> pubSubCommands;

    private final ObjectMapper objectMapper;

    @Inject
    public WithdrawRedisPubService(ReactiveRedisDataSource reactiveRedisDataSource, ObjectMapper objectMapper) {
        LOG.info("开始初始化Redis发布服务...");
        this.pubSubCommands = reactiveRedisDataSource.pubsub(String.class);
        this.objectMapper = objectMapper;
        if (this.pubSubCommands == null) {
            LOG.error("获取Redis PubSub命令对象失败！");
            return;
        }

        LOG.info("初始化Redis发布服务成功");
    }

    /**
     * 发布批量转账完成状态
     *
     * @return Uni<Void>表示发布操作的完成状态
     */
    public Uni<Void> publishBatchTransferStatus(List<UserWithdrawal> userWithdrawals) {
        try {
            if (this.pubSubCommands == null) {
                LOG.error("Redis PubSub命令对象为空，无法发布消息");
                return Uni.createFrom().failure(new IllegalStateException("Redis PubSub命令对象为空"));
            }
            String payload = objectMapper.writeValueAsString(userWithdrawals);
            LOG.infof("准备发布批量转账状态: %s 到频道: %s", payload, CacheNames.BATCH_TRANSFER_CHANNEL);
            // 直接返回Uni<Void>类型的结果，不进行转换
            return pubSubCommands.publish(CacheNames.BATCH_TRANSFER_CHANNEL, payload)
                    .onFailure().invoke(throwable -> {
                        LOG.error("发布批量转账状态失败", throwable);
                        System.err.println("发布批量转账状态失败: " + throwable.getMessage());
                    })
                    .onItem().invoke(() -> {
                        LOG.info("成功发布批量转账状态到频道: " + CacheNames.BATCH_TRANSFER_CHANNEL);
                    });
        } catch (Exception e) {
            LOG.error("发布Redis消息时发生异常", e);
            System.err.println("发布Redis消息时发生异常: " + e.getMessage());
            return Uni.createFrom().failure(e);
        }
    }

}
